package edu.ucsd.db.bassarisk.query.bindingtree.visitor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.tuple.Pair;

import edu.ucsd.db.bassarisk.algebra.logical.LogicalOperator;
import edu.ucsd.db.bassarisk.algebra.logical.LogicalPlanner;
import edu.ucsd.db.bassarisk.algebra.logical.Scan;
import edu.ucsd.db.bassarisk.algebra.logical.function.ColumnLength;
import edu.ucsd.db.bassarisk.algebra.logical.function.Function;
import edu.ucsd.db.bassarisk.columndef.DataColumnDefinition;
import edu.ucsd.db.bassarisk.exception.InsufficientDataException;
import edu.ucsd.db.bassarisk.exception.UnexpectedCaseException;
import edu.ucsd.db.bassarisk.query.ValidationTraversalLog;
import edu.ucsd.db.bassarisk.query.binding.BindingTuple;
import edu.ucsd.db.bassarisk.query.binding.Bindings.AggregatedValueBinding;
import edu.ucsd.db.bassarisk.query.binding.Bindings.Binding;
import edu.ucsd.db.bassarisk.query.binding.Bindings.BindingCombination;
import edu.ucsd.db.bassarisk.query.binding.Bindings.BindingUnion;
import edu.ucsd.db.bassarisk.query.binding.Bindings.ComputedValueBinding;
import edu.ucsd.db.bassarisk.query.binding.Bindings.MapFunctionValueBinding;
import edu.ucsd.db.bassarisk.query.binding.Bindings.PositionBinding;
import edu.ucsd.db.bassarisk.query.bindingtree.BindingTreeElements.BindingTreeEdge;
import edu.ucsd.db.bassarisk.query.bindingtree.BindingTreeElements.ChoiceNode;
import edu.ucsd.db.bassarisk.query.bindingtree.BindingTreeElements.ChoiceNode.Choice;
import edu.ucsd.db.bassarisk.query.bindingtree.BindingTreeElements.ChoiceNode.ChooseAll;
import edu.ucsd.db.bassarisk.query.bindingtree.BindingTreeElements.ChoiceNode.ChooseOne;
import edu.ucsd.db.bassarisk.query.bindingtree.BindingTreeElements.EstablishingNode;
import edu.ucsd.db.bassarisk.query.path.DSDElements.DSDDescent;
import edu.ucsd.db.bassarisk.schema.SchemaNode;
import edu.ucsd.db.bassarisk.treepattern.EdgeFunction;
import edu.ucsd.db.bassarisk.util.BooleanFlags;
import edu.ucsd.db.bassarisk.util.MultiMap;

public class BindingColumnsMappingTraversal
    implements
    BindingTreeVisitor<LogicalOperator, Pair<Boolean, Collection<DataColumnDefinition>>> {
  public enum MatchType {
    None,
    Superset,
    Exact
  }

  private final ValidationTraversalLog vtl;
  private final MultiMap<SchemaNode, DataColumnDefinition> schemaDataMap;
  private final Map<BindingTreeEdge, BindingTuple> bindingTuples;

  public BindingColumnsMappingTraversal(final MultiMap<SchemaNode, DataColumnDefinition> schemaDataMap,
                                        final Map<BindingTreeEdge, BindingTuple> bindingTuples,
                                        final ValidationTraversalLog vtl) {
    this.schemaDataMap = schemaDataMap;
    this.bindingTuples = bindingTuples;
    this.vtl = vtl;
  }

  @Override
  public LogicalOperator
      visitBindingTreeEdge(BindingTreeEdge edge,
                           Pair<Boolean, Collection<DataColumnDefinition>> args) {
    boolean isWithinConstraint = args.getLeft();
    Collection<DataColumnDefinition> candidateColumns = args.getRight();
    Collection<ComputedValueBinding> computations = edge.computations();
    List<AggregatedValueBinding> aggs = new ArrayList<AggregatedValueBinding>();
    List<MapFunctionValueBinding> maps = new ArrayList<MapFunctionValueBinding>();
    for (ComputedValueBinding computation : computations) {
      if (computation instanceof AggregatedValueBinding) {
        aggs.add((AggregatedValueBinding) computation);
      } else if (computation instanceof MapFunctionValueBinding) {
        maps.add((MapFunctionValueBinding) computation);
      } else {
        throw UnexpectedCaseException.ofSubtype(ComputedValueBinding.class);
      }
    }
    // if this is a single aggregate edge, then perhaps there is a view that can
    // will take care of it for us.
    if (aggs.size() == 1 && maps.size() == 0 && isWithinConstraint) {
      AggregatedValueBinding aggregateBinding = aggs.get(0);
      Binding targetBinding = aggregateBinding.getInputSequence();
      if (targetBinding.boundSequenceClass().getRefinement().size() == 1) {
        SchemaNode sequence =
            (SchemaNode) targetBinding.boundSequenceClass().getRefinement()
                .toArray()[0];
        for (DataColumnDefinition column : candidateColumns) {
          MatchType match =
              checkColumnSatisfaction(column, targetBinding, sequence,
                                      aggregateBinding.getFuntion());
          if (match == MatchType.Exact) { //
            String columnParam = "-";
            if (targetBinding instanceof PositionBinding) {
              columnParam = ((PositionBinding) targetBinding).getValueFilter();
            }
            return LogicalPlanner.createScan(column, columnParam, aggregateBinding);
          }
        }
      }
    }
    LogicalOperator subPlan =
        edge.tailNode().accept(this, Pair.of(isWithinConstraint, candidateColumns));
    if (computations.size() == 0) { return subPlan; }
    if (aggs.size() > 0) {
      // There is an optimization opportunity here. If we are just counting
      // an entire DB column, we can just get the answer from the metadata.
      if (!isWithinConstraint) {
        if (subPlan instanceof Scan) {
          Scan scanPlan = (Scan) subPlan;
          Function length =
              new ColumnLength(scanPlan.getColumn(), scanPlan.getBindingSatisfied());
          subPlan = LogicalPlanner.createConstantTuple(length);
        } else {
          subPlan = LogicalPlanner.createAggregateAll(aggs, subPlan);
        }
      } else {
        subPlan = LogicalPlanner.createAggregateWithin(aggs, subPlan);
      }
    }
    if (maps.size() > 0) {
      subPlan = LogicalPlanner.createMap(maps, subPlan);
    }
    if (edge.predicates().size() > 0) {
      subPlan = LogicalPlanner.createSelect(subPlan, edge.predicates());
    }
    return subPlan;
  }

  @Override
  public LogicalOperator
      visitChoiceNode(ChoiceNode choices,
                      Pair<Boolean, Collection<DataColumnDefinition>> args) {
    List<LogicalOperator> unionInputs = new ArrayList<LogicalOperator>();
    for (Choice choice : choices.getChoices()) {
      if (choice instanceof ChooseAll) {
        List<LogicalOperator> subPlans = new ArrayList<LogicalOperator>();
        for (BindingTreeEdge child : ((ChooseAll) choice).getChildren()) {
          subPlans.add(child.accept(this, args));
        }
        unionInputs.add(LogicalPlanner.createJoinWithin(subPlans));
      } else if (choice instanceof ChooseOne) {
        unionInputs.add(((ChooseOne) choice).getChild().accept(this, args));
      } else {
        throw UnexpectedCaseException.ofSubtype(Choice.class);
      }
    }
    return LogicalPlanner.createUnion(unionInputs);
  }

  @Override
  public LogicalOperator
      visitEstablishingNode(EstablishingNode node,
                            Pair<Boolean, Collection<DataColumnDefinition>> args) {
    // Each establishing node is either a terminal node in the binding tree (a
    // leaf) or it is a non-terminal (internal node). For now (it should be
    // different later) the non-terminal establishing nodes establish run-class
    // bindings, while terminal establishing nodes can establish either
    // run-class, attribute-class, or position-class bindings.
    // We handle terminal and non-terminal nodes differently.
    try {
      boolean isRoot =
          node.getEstablishedBinding().boundSequenceClass().getName()
              .equals("index");
      if (isRoot) { //
        return planRootNode(node, args.getRight());
      }
      Set<DataColumnDefinition> columnCandidates =
          new HashSet<DataColumnDefinition>();
      for (SchemaNode sequence : node.getEstablishedBinding().boundSequenceClass()
          .getRefinement()) {
        if (this.schemaDataMap.containsKey(sequence)) {
          columnCandidates.addAll(this.schemaDataMap.get(sequence));
        }
      }
      boolean isNonterminal = node.outDegree() > 0;
      if (isNonterminal) {
        return planNonterminalEstablishing(node, columnCandidates);
      } else {
        return planTerminalEstablishing(node, columnCandidates);
      }
    } catch (InsufficientDataException e) {
      vtl.addError(e);
    }
    return null;
  }

  private LogicalOperator
      planTerminalEstablishing(EstablishingNode node,
                               Collection<DataColumnDefinition> columnCandidates)
          throws InsufficientDataException {
    Binding binding = node.getEstablishedBinding();
    List<LogicalOperator> sequenceColumns = new ArrayList<LogicalOperator>();
    for (SchemaNode sequence : binding.boundSequenceClass().getRefinement()) {
      DataColumnDefinition sequenceColumn = null;
      columns: for (DataColumnDefinition candidateColumn : columnCandidates) {
        MatchType match =
            this.checkColumnSatisfaction(candidateColumn, binding, sequence,
                                         EdgeFunction.Values);
        switch (match) {
        case Exact:
          sequenceColumn = candidateColumn;
          break columns;
        default:
          break;
        }
      }
      if (sequenceColumn == null) {
        String message =
            "Database does not contain data for data sequence [%s] required by query.";
        message = String.format(message, sequence.toString());
        throw new InsufficientDataException(message);
      }
      String columnParam = "-";
      if (binding instanceof PositionBinding) {
        columnParam = ((PositionBinding) binding).getValueFilter();
      }
      sequenceColumns.add(LogicalPlanner.createScan(sequenceColumn, columnParam,
                                                    binding));
      sequenceColumn = null;
    }
    return LogicalPlanner.createUnion(sequenceColumns);
  }

  private LogicalOperator
      planRootNode(EstablishingNode node,
                   Collection<DataColumnDefinition> columnCandidates)
          throws InsufficientDataException {
    List<LogicalOperator> containedPlans = new ArrayList<LogicalOperator>();
    for (BindingTreeEdge childEdge : node.children()) {
      containedPlans.add(childEdge.accept(this, Pair.of(false, columnCandidates)));
    }
    return LogicalPlanner.createJoin(containedPlans);
  }

  private LogicalOperator
      planNonterminalEstablishing(EstablishingNode node,
                                  Collection<DataColumnDefinition> columnCandidates)
          throws InsufficientDataException {
    // First order of business is to determine the best source for the data
    // sequence bound to the established binding. The default is to simply
    // compute the union of the columns that store the data for each schema
    // node in the established bindings run-class. Typically, this is far
    // from optimal since there are probably views of these schema nodes
    // which pre-filter on some constraint. If there is a filter-implying
    // binding in some sub-tree that introduces the same (or a stronger)
    // constraint than the view, that view may be used as a smaller (faster)
    // source for the data sequence bound to the established binding.
    // .
    // As we search for these pre-filtered views, we may also find some
    // pre-aggregated value views, or pre-filtered attribute views. We
    // should also collect these where they are useful.
    // .
    // Check each child edge, and each binding that originates from that child
    // edge, to see if there is a view from which we can get the data from
    // directly. Things to think about:
    // 1) The cardinality of a binding incident to this establishing node.
    // .... a) A filtering binding must always be a singleton - a single value
    // ....... "contains" or "not-contains".
    // .... b) An aggregate is necessarily a singleton if it aggregates in the
    // ....... context of this establishing node.
    // .... c) Any singleton value can potentially be pulled in directly from
    // ....... a view.
    // .... d) A child edge from this node may have both singleton filtering
    // ....... bindings and non-singleton value bindings at the same time. The
    // ....... filter bindings may still be used to pre-filter the binding
    // ....... established at this node.
    // 2) A materialized data column may fully satisfy a filtering binding in
    // .. the context of this binding node, or only partially satisfy it.
    // .. A data column partially satisfies a considered binding when:
    // .... a) The data column is grouped on a run-class that is a super set
    // ....... of the run-class for this established binding.
    // .... b) The data column leaf class is a super-set of the sequence-class
    // ....... established by the binding considered.
    // .. Furthermore, the data-column fully satisfies when:
    // .....c) The binding considered is not an input to any functions or
    // ....... predicates and is not an output binding.
    // 3) A child edge that only provides bindings that are fully satisfied by
    // .. views does not need to be made a sub-plan.
    // .
    // First, find the set of columns which define aggregate on on the run-class
    // of this established binding. These columns will be the candidates from
    // which matching columns are drawn.
    List<LogicalOperator> containedPlans = new ArrayList<LogicalOperator>();
    List<LogicalOperator> containerPlans = new ArrayList<LogicalOperator>();
    for (BindingTreeEdge childEdge : node.children()) {
      if (columnCandidates.size() == 0) {
        // shortcut: there are no potential views anyway, build the simple plan
        containedPlans.add(childEdge.accept(this, Pair.of(true, columnCandidates)));
        continue;
      }
      if (!childEdge.functions().contains(EdgeFunction.NonEmpty)) {
        // this is not a filtering child, so searching for pre-filters is futile
        containedPlans.add(childEdge.accept(this, Pair.of(true, columnCandidates)));
        continue;
      }
      BindingTuple tuple = this.bindingTuples.get(childEdge);
      BooleanFlags directPlan = new BooleanFlags(tuple.size(), false);
      BooleanFlags filterPlan = new BooleanFlags(tuple.size(), false);
      int bindingIndex = 0;
      List<LogicalOperator> bindingPlans = new ArrayList<LogicalOperator>();
      List<Binding> filterCandidates = new ArrayList<Binding>();
      Set<Binding> topLevelFilterCandidates = new HashSet<Binding>();
      filterCandidates.addAll((Collection<? extends Binding>) tuple.bindings());
      for (Binding binding : filterCandidates) {
        if (binding instanceof ComputedValueBinding) {
          continue; // skip it
        }
        while (filterCandidates.contains(binding.getParent())) {
          binding = binding.getParent();
        }
        topLevelFilterCandidates.add(binding);
      }
      for (Binding binding : topLevelFilterCandidates) {
        MatchType result = // TODO
            planBinding(node.getEstablishedBinding(), binding, childEdge,
                        columnCandidates, bindingPlans, EdgeFunction.NonEmpty);
        switch (result) {
        case Exact:
          directPlan.makeTrue(bindingIndex);
        case Superset:
          filterPlan.makeTrue(bindingIndex);
          break;
        default:
          // do nothing
          break;
        }
        bindingIndex++;
      }
      if (directPlan.allTrue()) {
        // then the cheapest plan will intersect these results
        containerPlans.addAll(bindingPlans);
      } else if (!filterPlan.allFalse()) {
        // find the most effective (lowest cost vs strongest) filter
        // intersection of these filters
        containerPlans.addAll(bindingPlans);
        // FIXME cost planning here, not all those filters should be included.
        containedPlans.add(childEdge.accept(this, Pair.of(true, columnCandidates)));
      } else {
        // find the lowest cost plan recursively without filters
        containedPlans.add(childEdge.accept(this, Pair.of(true, columnCandidates)));
      }
    }
    // If any filter plan was generated, then that will produce the required
    // container tuples. Else, we create a plan here to produce the container
    // tuples.
    if (containerPlans.size() == 0) {
      containerPlans.add(planTerminalEstablishing(node, columnCandidates));
    }
    LogicalOperator intersection = LogicalPlanner.createIntersection(containerPlans);
    if (containedPlans.size() == 0) {
      return intersection;
    } else {
      return LogicalPlanner.createWithinContext(intersection, LogicalPlanner
          .createJoinWithin(containedPlans));
    }
  }

  private MatchType planBinding(Binding satBinding, Binding binding,
                                BindingTreeEdge childEdge,
                                Collection<DataColumnDefinition> columnCandidates,
                                List<LogicalOperator> filterPlans,
                                EdgeFunction function) {
    if (binding instanceof BindingCombination) {
      List<LogicalOperator> subFilters = new ArrayList<LogicalOperator>();
      // FIXME we are just assuming that everything finds a column. naughty
      // naughty
      for (Binding child : ((BindingCombination) binding).getCombinedBindings()) {
        planBinding(binding, child, childEdge, columnCandidates, subFilters, function);
      }
      if (subFilters.size() == 0) { return MatchType.None; }
      if (binding instanceof BindingUnion) {
        filterPlans.add(LogicalPlanner.createUnion(subFilters));
        return MatchType.Exact;
      } else {
        filterPlans.add(LogicalPlanner.createIntersection(subFilters));
        return MatchType.Exact;
      }
    }
    Collection<? extends SchemaNode> refinement =
        binding.boundSequenceClass().getRefinement();
    int refinementSize = refinement.size();
    BooleanFlags exactMatch = new BooleanFlags(refinementSize, false);
    BooleanFlags supersetMatch = new BooleanFlags(refinementSize, false);
    List<LogicalOperator> bindingPlans = new ArrayList<LogicalOperator>();
    for (SchemaNode sequence : refinement) {
      DataColumnDefinition bestDef = null;
      int columnIndex = 0;
      columns: for (DataColumnDefinition candidateColumn : columnCandidates) {
        if (candidateColumn.aggregationFunction() != function) {
          continue;
        }
        DSDDescent aggregationContext = candidateColumn.getAggregationContext();
        if (!aggregationContext.hasNext()) {
          continue;
        }
        MatchType match =
            checkColumnSatisfaction(candidateColumn, binding, sequence,
                                    EdgeFunction.NonEmpty);
        switch (match) {
        case Exact:
          bestDef = candidateColumn;
          exactMatch.makeTrue(columnIndex);
          supersetMatch.makeTrue(columnIndex);
          break columns;
        case None:
          break;
        case Superset:
          if (bestDef == null || bestDef.cost() > candidateColumn.cost()) {
            bestDef = candidateColumn;
            supersetMatch.makeTrue(columnIndex);
          }
          break;
        default:
          throw UnexpectedCaseException.ofEnum(MatchType.class);
        }
        columnIndex++;
      }
      String columnParam = "-";
      if (binding instanceof PositionBinding) {
        columnParam = ((PositionBinding) binding).getValueFilter();
      }
      bindingPlans.add(LogicalPlanner.createScan(bestDef, columnParam, satBinding));
    }
    if (exactMatch.allTrue()) {
      filterPlans.add(LogicalPlanner.createUnion(bindingPlans));
      return MatchType.Exact;
    } else if (supersetMatch.allTrue()
        && childEdge.functions().contains(EdgeFunction.NonEmpty)) {
      filterPlans.add(LogicalPlanner.createUnion(bindingPlans));
      return MatchType.Superset;
    }
    return MatchType.None;
  }

  private MatchType checkColumnSatisfaction(DataColumnDefinition column,
                                            Binding binding,
                                            SchemaNode boundSequence,
                                            EdgeFunction function) {
    // The first order of business is to figure out what relationship the
    // binding has to the operator context (childEdge)
    if (function == null) { //
      return MatchType.None; // Not a relationship described by our views.
    }
    if (function != column.aggregationFunction()) { //
      return MatchType.None; // Column performs a different aggregation.
    }
    // We need the target extracted or aggregated data sequence to match
    if (boundSequence != column.getTargetSequence()) { //
      return MatchType.None;
    }
    // FIXME we finally need to check for proper structure constraints.
    return MatchType.Exact;
  }
}
